package com.zhangsr.zk.client;

import com.zhangsr.zk.client.exception.ZkException;
import com.zhangsr.zk.client.exception.ZkInterruptedException;
import com.zhangsr.zk.client.listener.ZkClientListener;
import com.zhangsr.zk.client.serialize.SerializableSerializer;
import com.zhangsr.zk.client.serialize.ZkSerializer;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;

import java.util.concurrent.Callable;

/**
 * @Author: zhangsr
 */
public class ZkClient extends ZkClientListener implements Watcher {

    private final static Logger log = Logger.getLogger(ZkClient.class);

    private IZkConnection connection;

    public ZkClient(String zkServers, int sessionTimeout) {
        this(new ZkConnection(zkServers, sessionTimeout), new SerializableSerializer());
    }

    private ZkClient(IZkConnection connection, ZkSerializer zkSerializer) {
        this.connection = connection;
        super.zkSerializer = zkSerializer;
        connect(this);
    }

    private void connect(Watcher watcher) throws IllegalStateException {
        log.info("try reconnect ...");
        try {
            initZkEventThread();
            connection.connect(watcher);
        } catch (Exception e) {
            log.error("connect error!", e);
        }
    }

    public void close() throws InterruptedException {
        if (connection == null) {
            return;
        }
        log.info("Closing ZkClient ..");
        try {
            super.eventThread.interrupt();
            super.eventThread.join(2000);
            connection.close();
            connection = null;
        } catch (ZkInterruptedException e) {
            throw new ZkInterruptedException(e);
        }
    }

    public void reconnect() {
        log.warn("try reconnect ...");
        try {
            connection.close();
            connection.connect(this);
        } catch (Exception e) {
            log.error("reconnect error!", e);
        }
    }

    /**
     * 处理节点事件
     *
     * @param watchedEvent
     */
    public void process(WatchedEvent watchedEvent) {
        log.info("Received event : " + watchedEvent);
        boolean stateChanged = watchedEvent.getPath() == null;
        boolean znodeChanged = watchedEvent.getPath() != null;
        boolean dataChanged = watchedEvent.getType() == Watcher.Event.EventType.NodeDataChanged
                || watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted
                || watchedEvent.getType() == Watcher.Event.EventType.NodeCreated
                || watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged;

        try {
            if (stateChanged) {
                processConnectionChanged(watchedEvent);
            }
            if (dataChanged) {
                processDataOrChildChange(watchedEvent);
            }
        } finally {

        }
    }

    /**
     * 创建节点
     *
     * @param path
     * @param data
     * @param mode
     * @return
     * @throws Exception
     */
    public String create(final String path, Object data, final CreateMode mode) throws Exception {
        if (path == null) {
            throw new NullPointerException("path must not be null.");
        }
        final byte[] bytes = data == null ? null : serialize(data);

        return retryUntilConnected(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return connection.create(path, bytes, mode);
            }
        });
    }

    /**
     * 判断节点为是否存在
     *
     * @param path
     * @return
     */
    public boolean exists(String path) {
        return exists(path, hasListeners(path));
    }

    public boolean exists(final String path, final boolean watch) {
        return retryUntilConnected(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return connection.exists(path, watch);
            }
        });
    }

    /**
     * 取节点数据
     *
     * @param path
     * @param <T>
     * @return
     */
    public <T extends Object> T readData(String path) {
        return readData(path, false);
    }

    public <T extends Object> T readData(String path, boolean returnNullIfPathNotExists) {
        T data = null;
        try {
            data = readData(path, null);
        } catch (Exception e) {

        }
        return data;
    }

    public <T extends Object> T readData(String path, Stat stat) {
        return readData(path, stat, hasListeners(path));
    }

    public <T extends Object> T readData(final String path, final Stat stat, final boolean watch) {
        byte[] data = retryUntilConnected(new Callable<byte[]>() {
            @Override
            public byte[] call() throws Exception {
                return connection.readData(path, stat, watch);
            }
        });
        return derializable(data);
    }

    private <T> T retryUntilConnected(Callable<T> callable) throws IllegalArgumentException, ZkException, RuntimeException {
        while (true) {
            try {
                return callable.call();
            } catch (Exception e) {

            }
        }
    }

}
